package com.gph.saviorframework.mq.api.impl;

import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.gph.saviorframework.mq.api.*;
import com.gph.saviorframework.mq.api.impl.rocketmq.*;
import com.gph.saviorframework.mq.api.impl.util.MQUtil;
import com.gph.saviorframework.mq.api.order.OrderConsumer;
import com.gph.saviorframework.mq.api.order.OrderProducer;
import com.gph.saviorframework.mq.api.transaction.LocalTransactionChecker;
import com.gph.saviorframework.mq.api.transaction.TransactionProducer;
import com.gph.saviorframework.mq.api.transaction.TransactionStatus;

import java.util.Properties;

/**
 * Created by root on 16-7-10.
 */
public class MQFactoryImpl implements MQFactoryAPI{

    public Producer createProducer(Properties properties) {
        ProducerImpl producer = new ProducerImpl(properties);
        return producer;
    }

    public Consumer createConsumer(Properties properties) {
        ConsumerImpl consumer = new ConsumerImpl(properties);
        return consumer;
    }

    /**
     * 创建顺序消息生产者
     *
     * @param properties
     * @return
     */
    public OrderProducer createOrderProducer(Properties properties) {
        OrderProducerImpl producer = new OrderProducerImpl(properties);
        return producer;
    }

    /**
     * 创建顺序消息消费者
     *
     * @param properties
     * @return
     */
    public OrderConsumer createOrderedConsumer(Properties properties) {
        OrderConsumerImpl consumer = new OrderConsumerImpl(properties);
        return consumer;
    }

    /**
     * 创建事务处理的消息生产者
     *
     * @param properties
     * @return
     */
    public TransactionProducer createTransactionProducer(Properties properties,
                                                         final LocalTransactionChecker checker) {
        TransactionProducerImpl transactionProducer = new TransactionProducerImpl(properties, new TransactionCheckListener() {
            public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
                String msgId = msg.getProperty(Constants.TRANSACTION_ID);
                Message message = MQUtil.msgConvert(msg);
                message.setMsgID(msgId);
                TransactionStatus check = checker.check(message);
                if (TransactionStatus.CommitTransaction == check) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                else if (TransactionStatus.RollbackTransaction == check) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });
        return transactionProducer;
    }
}
